package com.mask.im.server.listener;

import com.alibaba.fastjson.JSON;
import com.mask.im.sdk.dto.IMPushMessage;
import com.mask.im.server.push.IMPushMessageFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 消息消费者，处理不同类型的消息
 *
 * @author jian
 */
@Slf4j
@Component
public class MessageConsumer {

    @Resource
    private IMPushMessageFactory IMPushMessageFactory;

    @Value("${im.kafka.topic:im-messages}")
    private String kafkaTopic;


    @KafkaListener(topics = "${im.kafka.topic:im-messages}", containerFactory = "kafkaListenerContainerFactory")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            // 检查消息内容是否为空
            if (record.value() == null || record.value().trim().isEmpty()) {
                log.warn("收到空消息，跳过处理: {}", record);
                acknowledgment.acknowledge();
                return;
            }

            // 解析消息
            log.info("MessageConsumer: {}", record);
            IMPushMessage message = JSON.parseObject(record.value(), IMPushMessage.class);
            log.info("收到消息: 类型={}, 发送者={}, 接收者={}",
                    message.getPushType(),
                    message.getSenderId(),
                    message.getReceiverId());
            Boolean success = IMPushMessageFactory.pushMessage(message);
            // 处理成功则提交offset
            if (success) {
                acknowledgment.acknowledge();
                log.info("消息处理成功，ID: {}", message.getContent());
            } else {
                log.error("消息处理失败，ID: {}", message.getContent());
            }
        } catch (Exception e) {
            log.error("消息消费异常", e);
            // 异常时不提交offset，等待重试
        }
    }
}
    